package com.wande.dataplatform.filecollection.parser.impl;

import cn.hutool.core.util.StrUtil;
import com.fasterxml.jackson.core.JsonParser;
import com.fasterxml.jackson.core.JsonToken;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.jayway.jsonpath.Configuration;
import com.jayway.jsonpath.JsonPath;
import com.jayway.jsonpath.Option;
import com.wande.dataplatform.filecollection.common.enums.FileCollectionErrorCode;
import com.wande.dataplatform.filecollection.common.exception.FileCollectionException;
import com.wande.dataplatform.filecollection.domain.dto.FieldInfo;
import com.wande.dataplatform.filecollection.parser.AbstractFileParser;
import com.wande.dataplatform.filecollection.parser.ParserConfig;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import java.io.*;
import java.util.*;
import java.util.function.Consumer;

/**
 * JSON文件解析器
 * 支持JSONPath表达式提取数据
 * 支持嵌套对象和数组处理
 * 支持流式解析
 *
 * @author dataplatform
 * @date 2025-01-24
 */
@Slf4j
@Component
public class JsonFileParser extends AbstractFileParser {

    private static final int SAMPLE_SIZE = 100;
    private static final String DEFAULT_DATA_PATH = "$";
    private final ObjectMapper objectMapper = new ObjectMapper();

    @Override
    public String getSupportedFileType() {
        return "JSON";
    }

    @Override
    public List<FieldInfo> detectFields(InputStream inputStream, ParserConfig config) {
        validateInputStream(inputStream);
        validateConfig(config);

        List<FieldInfo> fields = new ArrayList<>();

        try {
            String jsonContent = new String(inputStream.readAllBytes(), config.getCharset());
            String dataPath = getDataPath(config);

            Configuration conf = Configuration.builder()
                    .options(Option.DEFAULT_PATH_LEAF_TO_NULL, Option.SUPPRESS_EXCEPTIONS)
                    .build();

            Object result = JsonPath.using(conf).parse(jsonContent).read(dataPath);

            if (result == null) {
                return fields;
            }

            List<Map<String, Object>> records = extractRecords(result);
            if (records.isEmpty()) {
                return fields;
            }

            // 收集所有字段名和样本数据
            Map<String, List<String>> fieldSamples = new LinkedHashMap<>();
            int sampleCount = 0;

            for (Map<String, Object> record : records) {
                if (sampleCount >= SAMPLE_SIZE) {
                    break;
                }

                flattenMap(record, "", fieldSamples);
                sampleCount++;
            }

            // 创建字段信息
            int index = 0;
            for (Map.Entry<String, List<String>> entry : fieldSamples.entrySet()) {
                String fieldName = cleanFieldName(entry.getKey());
                List<String> samples = entry.getValue();
                String fieldType = inferFieldTypeFromSamples(samples);

                FieldInfo fieldInfo = createFieldInfo(fieldName, fieldType, index);
                if (!samples.isEmpty()) {
                    fieldInfo.setSampleValue(samples.get(0));
                }
                fields.add(fieldInfo);
                index++;
            }

        } catch (IOException e) {
            log.error("Failed to detect JSON fields", e);
            throw new FileCollectionException(FileCollectionErrorCode.FILE_PARSE_ERROR, "Failed to detect JSON fields: " + e.getMessage());
        }

        return fields;
    }

    @Override
    public Iterator<Map<String, Object>> parse(InputStream inputStream, ParserConfig config) {
        validateInputStream(inputStream);
        validateConfig(config);

        try {
            String jsonContent = new String(inputStream.readAllBytes(), config.getCharset());
            String dataPath = getDataPath(config);

            Configuration conf = Configuration.builder()
                    .options(Option.DEFAULT_PATH_LEAF_TO_NULL, Option.SUPPRESS_EXCEPTIONS)
                    .build();

            Object result = JsonPath.using(conf).parse(jsonContent).read(dataPath);
            List<Map<String, Object>> records = extractRecords(result);

            return new JsonIterator(records, config);

        } catch (IOException e) {
            log.error("Failed to parse JSON file", e);
            throw new FileCollectionException(FileCollectionErrorCode.FILE_PARSE_ERROR, "Failed to parse JSON file: " + e.getMessage());
        }
    }

    @Override
    public void parseStream(InputStream inputStream, ParserConfig config, Consumer<Map<String, Object>> consumer) {
        validateInputStream(inputStream);
        validateConfig(config);

        try {
            // 对于大JSON文件，使用流式解析
            JsonParser parser = objectMapper.getFactory().createParser(inputStream);

            // 如果是数组，逐个解析元素
            if (parser.nextToken() == JsonToken.START_ARRAY) {
                while (parser.nextToken() != JsonToken.END_ARRAY) {
                    JsonNode node = objectMapper.readTree(parser);
                    Map<String, Object> rowData = flattenJsonNode(node, "", config);

                    // 跳过空行
                    if (config.getSkipEmptyRows() && isEmptyRow(rowData)) {
                        continue;
                    }

                    consumer.accept(rowData);
                }
            } else {
                // 单个对象
                JsonNode node = objectMapper.readTree(parser);
                Map<String, Object> rowData = flattenJsonNode(node, "", config);

                if (!config.getSkipEmptyRows() || !isEmptyRow(rowData)) {
                    consumer.accept(rowData);
                }
            }

        } catch (IOException e) {
            log.error("Failed to parse JSON file in stream mode", e);
            throw new FileCollectionException(FileCollectionErrorCode.FILE_PARSE_ERROR, "Failed to parse JSON file in stream mode: " + e.getMessage());
        }
    }

    @Override
    public boolean validate(InputStream inputStream, ParserConfig config) {
        try {
            objectMapper.readTree(inputStream);
            return true;
        } catch (Exception e) {
            log.warn("JSON validation failed: {}", e.getMessage());
            return false;
        }
    }

    /**
     * 获取数据路径
     *
     * @param config 解析器配置
     * @return JSONPath表达式
     */
    private String getDataPath(ParserConfig config) {
        if (StrUtil.isNotBlank(config.getJsonDataPath())) {
            return config.getJsonDataPath();
        }
        if (StrUtil.isNotBlank(config.getJsonRootPath())) {
            return config.getJsonRootPath();
        }
        return DEFAULT_DATA_PATH;
    }

    /**
     * 提取记录列表
     *
     * @param result JSONPath查询结果
     * @return 记录列表
     */
    @SuppressWarnings("unchecked")
    private List<Map<String, Object>> extractRecords(Object result) {
        List<Map<String, Object>> records = new ArrayList<>();

        if (result instanceof List) {
            List<?> list = (List<?>) result;
            for (Object item : list) {
                if (item instanceof Map) {
                    records.add((Map<String, Object>) item);
                } else {
                    // 基本类型，包装成Map
                    Map<String, Object> record = new LinkedHashMap<>();
                    record.put("value", item);
                    records.add(record);
                }
            }
        } else if (result instanceof Map) {
            records.add((Map<String, Object>) result);
        } else {
            // 基本类型，包装成Map
            Map<String, Object> record = new LinkedHashMap<>();
            record.put("value", result);
            records.add(record);
        }

        return records;
    }

    /**
     * 展平嵌套Map（用于字段检测）
     *
     * @param map 嵌套Map
     * @param prefix 字段名前缀
     * @param fieldSamples 字段样本数据
     */
    @SuppressWarnings("unchecked")
    private void flattenMap(Map<String, Object> map, String prefix, Map<String, List<String>> fieldSamples) {
        for (Map.Entry<String, Object> entry : map.entrySet()) {
            String key = entry.getKey();
            Object value = entry.getValue();
            String fieldName = StrUtil.isBlank(prefix) ? key : prefix + "_" + key;

            if (value == null) {
                fieldSamples.computeIfAbsent(fieldName, k -> new ArrayList<>()).add("");
            } else if (value instanceof Map) {
                flattenMap((Map<String, Object>) value, fieldName, fieldSamples);
            } else if (value instanceof List) {
                List<?> list = (List<?>) value;
                if (!list.isEmpty() && list.get(0) instanceof Map) {
                    // 数组的第一个元素是对象，递归展平
                    flattenMap((Map<String, Object>) list.get(0), fieldName, fieldSamples);
                } else {
                    // 基本类型数组，转换为字符串
                    fieldSamples.computeIfAbsent(fieldName, k -> new ArrayList<>()).add(value.toString());
                }
            } else {
                fieldSamples.computeIfAbsent(fieldName, k -> new ArrayList<>()).add(value.toString());
            }
        }
    }

    /**
     * 展平JsonNode
     *
     * @param node JsonNode
     * @param prefix 字段名前缀
     * @param config 解析器配置
     * @return 展平后的Map
     */
    private Map<String, Object> flattenJsonNode(JsonNode node, String prefix, ParserConfig config) {
        Map<String, Object> result = new LinkedHashMap<>();

        if (node.isObject()) {
            Iterator<Map.Entry<String, JsonNode>> fields = node.fields();
            while (fields.hasNext()) {
                Map.Entry<String, JsonNode> field = fields.next();
                String key = field.getKey();
                JsonNode value = field.getValue();
                String fieldName = StrUtil.isBlank(prefix) ? key : prefix + "_" + key;

                if (value.isObject() || value.isArray()) {
                    result.putAll(flattenJsonNode(value, fieldName, config));
                } else {
                    String strValue = value.isNull() ? "" : value.asText();
                    strValue = processFieldValue(strValue, config);
                    result.put(cleanFieldName(fieldName), strValue);
                }
            }
        } else if (node.isArray()) {
            for (int i = 0; i < node.size(); i++) {
                JsonNode item = node.get(i);
                String fieldName = prefix + "_" + i;
                result.putAll(flattenJsonNode(item, fieldName, config));
            }
        } else {
            String strValue = node.isNull() ? "" : node.asText();
            strValue = processFieldValue(strValue, config);
            result.put(cleanFieldName(prefix), strValue);
        }

        return result;
    }

    /**
     * JSON迭代器
     */
    private class JsonIterator implements Iterator<Map<String, Object>> {
        private final List<Map<String, Object>> records;
        private final ParserConfig config;
        private int currentIndex;

        public JsonIterator(List<Map<String, Object>> records, ParserConfig config) {
            this.records = records;
            this.config = config;
            this.currentIndex = 0;
        }

        @Override
        public boolean hasNext() {
            return currentIndex < records.size();
        }

        @Override
        public Map<String, Object> next() {
            if (!hasNext()) {
                throw new NoSuchElementException();
            }

            Map<String, Object> record = records.get(currentIndex);
            currentIndex++;

            // 展平嵌套结构
            Map<String, Object> flatRecord = new LinkedHashMap<>();
            flattenForOutput(record, "", flatRecord, config);

            return flatRecord;
        }

        @SuppressWarnings("unchecked")
        private void flattenForOutput(Map<String, Object> map, String prefix, 
                                     Map<String, Object> output, ParserConfig config) {
            for (Map.Entry<String, Object> entry : map.entrySet()) {
                String key = entry.getKey();
                Object value = entry.getValue();
                String fieldName = StrUtil.isBlank(prefix) ? key : prefix + "_" + key;

                if (value instanceof Map) {
                    flattenForOutput((Map<String, Object>) value, fieldName, output, config);
                } else if (value instanceof List) {
                    output.put(cleanFieldName(fieldName), value.toString());
                } else {
                    String strValue = value == null ? "" : value.toString();
                    strValue = processFieldValue(strValue, config);
                    output.put(cleanFieldName(fieldName), strValue);
                }
            }
        }
    }
}
